﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using NetMQ;

namespace NETMQLib
{
    /// <summary>
    /// 订阅MQ
    /// </summary>
    public class SubMQ
    {
        NetMQ.Sockets.SubscriberSocket socket = null;
        public event Action<string> ReceiveFrameStringEvent;
        public BaseMQModel BaseMQInfo { get; set; }

        string _topic = string.Empty;
        public SubMQ(string ip, int port, string topic = "")
        {
            BaseMQInfo = new BaseMQModel() { IP = ip, Port = port };
            string url = $"tcp://{ip}:{port}";
            _topic = topic;
            socket = new NetMQ.Sockets.SubscriberSocket();
            socket.Connect(url);
            socket.Subscribe(topic);
            socket.ReceiveReady += Socket_ReceiveReady;
            Task.Factory.StartNew(() =>
            {
                while (true)
                socket.Poll(TimeSpan.FromMilliseconds(10));
            });
        }

        private void Socket_ReceiveReady(object sender, NetMQSocketEventArgs e)
        {
            try
            {
                string topic = e.Socket.ReceiveFrameString();
                string msg = e.Socket.ReceiveFrameString();
                if (topic != _topic) 
                    return;

                ReceiveFrameStringEvent?.Invoke(msg);
            }
            catch (Exception ex)
            {
            }
        }


        public void Dispose()
        {
            if (socket != null && !socket.IsDisposed)
            {
                socket.Dispose();
                socket.Close();
            }
        }
    }
}
